package com.alpaca.recommend.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;

/**
 * 获取连接,使用时注意关闭hTable的连接
 * Created by wangj on 2015/1/28.
 */
@Service
public class HTableClients {
    private Logger logger = LoggerFactory.getLogger(HTableClients.class);

    private Connection connection;

    private String clientPort = "2181";
    private String zkAddr = "121.40.66.213,121.40.127.216,121.41.73.63";
//    private String zkAddr = "10.5.117.109";


    public HTableClients()  {
        try {
            Configuration configuration = HBaseConfiguration.create();
            configuration.set("hbase.zookeeper.property.clientPort", clientPort);
            configuration.set("hbase.zookeeper.quorum", zkAddr);
            connection = ConnectionFactory.createConnection(configuration);
        }catch (IOException e) {
            e.printStackTrace();
        }
    }


    public Configuration getConf(){
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", clientPort);
        configuration.set("hbase.zookeeper.quorum", zkAddr);
        return configuration;
    }

    @PreDestroy
    public void close() {
        try {
            logger.info("close hbase connect!");
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取htable实例
     * @param tableName
     * @return
     * @throws java.io.IOException
     */
    public Table getHTable(String tableName) throws IOException {
        return connection.getTable(TableName.valueOf(tableName.getBytes()));
    }

    public String getZkAddr() {
        return zkAddr;
    }

    public void setZkAddr(String zkAddr) {
        this.zkAddr = zkAddr;
    }

    public String getClientPort() {
        return clientPort;
    }

    public void setClientPort(String clientPort) {
        this.clientPort = clientPort;
    }
}
